package com.cnsuning.jobtool.jobtools.transfer;

import org.apache.commons.lang3.StringUtils;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

public class Transfer {

	/** 异常数据保存目录 */
	private static String filePath = "E:\\pre\\error.txt";
	private static PreparedStatement preparedStatement;
	private static ResultSet resultSet;



	/** 每次从源库中抽取数据量 */
	private static int dataCount = 5000;

	public static void main(String[] args) {
        String createdBy = "CREATED_TIME" ;
		// 将业务库源库中数据迁移到目标库中   CREAT_TIME   CREATE_TIME   CREATED_TIME
		List<String> tableList = Utils.getTableName();
		for(String table : tableList){
			System.out.println("开始迁移数据表:"+table);
			transferDataY("xgpre", "xgpre", table, dataCount, "omsd",createdBy);
            //deletedataformysql( "omsdpremycat", table,"CREATE_ID");
			System.out.println("迁移数据完成表:"+table);
		}
        System.out.println("本次所有表迁移完==========>");
	}

	//删除目标库数据
    public static void deletedataformysql( String envByTo, String tableName, String creatByOrCreatId) {

        // 获取目标库连接(myCat连接)
        ConnectionInfo myCatConnectionInfo = Utils.getNewMySqlConnectionInfo(envByTo);
        Connection toConnection = Utils.getConnection(myCatConnectionInfo.getDriver(), myCatConnectionInfo.getUrl(), myCatConnectionInfo.getUserName(),
                myCatConnectionInfo.getPassword());
        StringBuilder sql_delete = new StringBuilder("delete from ");
        sql_delete.append(tableName).append(" where ").append(creatByOrCreatId).append(" != 'OMSD_M';");
        System.out.println("sql:  "+sql_delete.toString());
        //删除数据
        int result = 0;
        try {
            result = executeUpdate(toConnection, sql_delete.toString());
        } catch (SQLException e) {
            System.out.println("删除" + tableName + "失败 : " + e);
        }
        System.out.println("删除" + tableName + "mysql数据结果 : " + result);

    }
	/**
	 * 将业务库源库中数据迁移到目标库中
	 *
	 * @param envByFrom
	 * @param envByTo
	 * @param tableName
	 * @param batchCount
	 * @param dbType
	 */
	private static void transferDataY(String envByFrom, String envByTo, String tableName, int batchCount,
			String dbType,String creatTime) {

		long start = System.currentTimeMillis();
		// 获取源库连接信息
		List<ConnectionInfo> fromConnectionInfoList = Utils.getFromConnectionInfo(envByFrom, dbType);
		// 获取目标库连接(myCat连接)
		ConnectionInfo myCatConnectionInfo = Utils.getNewMySqlConnectionInfo(envByTo);
		Connection toConnection = Utils.getConnection(myCatConnectionInfo.getDriver(), myCatConnectionInfo.getUrl(), myCatConnectionInfo.getUserName(),
				myCatConnectionInfo.getPassword());

		for (ConnectionInfo fromConnectionInfo : fromConnectionInfoList) {
			Connection fromConnection = Utils.getConnection(fromConnectionInfo.getDriver(), fromConnectionInfo.getUrl(),
					fromConnectionInfo.getUserName(), fromConnectionInfo.getPassword());
			String fields = "";
			// 1.获取源表除主键ROW_ID之外的字段
			try {
                System.out.println("当前操作的数据库是: "+fromConnectionInfo.getUserName());
				fields = getFields(fromConnection, tableName.toUpperCase(), fromConnectionInfo.getScheam());
			} catch (SQLException e) {
				System.out.println("查询源表字段时发生异常: {库名: "
						+ fromConnectionInfo.getUrl().substring(fromConnectionInfo.getUrl().lastIndexOf("/") + 1)
						+ " 表名: " + tableName.toUpperCase() + "}, 异常信息: {" + e
						+ "}");
			}
			int count = 0;
		while (true) {
			    int startIndex = batchCount*count;
				int endIndex = batchCount*(count+1);
				StringBuilder fromSql = new StringBuilder();
		        /*fromSql.append("SELECT ");
				fromSql.append(fields);
				fromSql.append(" FROM ");
				fromSql.append("gomstbiz."+tableName.toUpperCase());
				fromSql.append(" WHERE CREATED_TIME >= '2018-01-23 00:00:00' AND CREATED_TIME <= '2018-01-31 23:59:59'");*/
				fromSql.append("   SELECT page_row_.* FROM(SELECT ROWNUMBER() OVER() AS rowid_,orgin_tbl_.* FROM(SELECT ");
				fromSql.append(fields);
				fromSql.append(" FROM ");
				fromSql.append(fromConnectionInfo.getScheam()+"."+tableName.toUpperCase());
//				String sqlCondition=" WHERE "+creatTime+" >= '2019-06-01 00:00:00' AND "+creatTime+" <= '2019-12-23 00:00:00'";
//				fromSql.append(sqlCondition);
				fromSql.append(" )orgin_tbl_) page_row_ WHERE page_row_.rowid_ >  ");
				fromSql.append(startIndex);
				fromSql.append(" AND page_row_.rowid_ <= ");
				fromSql.append(endIndex);

				List<String> fromList = null;
				long time  = System.currentTimeMillis();
				try {
					fromList = getFromData(fromConnection, fromSql.toString(),
							tableName.toUpperCase());
				} catch (SQLException e) {
					System.out.println("查询源表数据时发生异常: {库名: "
							+ fromConnectionInfo.getUrl().substring(fromConnectionInfo.getUrl().lastIndexOf("/"))
							+ " 表名: " + tableName.toUpperCase());
					System.out.println(""+e);
				}
				System.out.println("查询数据所用时间"+ (System.currentTimeMillis()-time));
				try {
					if (fromList != null && fromList.size() > 0) {
                        System.out.println("db2源库实际条数: "+fromList.size());
						saveDataBatch(toConnection, tableName, fields, batchCount, fromList);
					}
				} catch (SQLException e) {
					System.out.println("保存数据时发生异常: {" + e + "}");

				}
				if (fromList == null || fromList.size() < dataCount) {
				   break;
			   }
				count++;

			}

		}
		long end = System.currentTimeMillis();
		System.out.println("总共耗时: " + (end - start) / 1000 + "秒");
	}

	/**
	 * 查询表字段，不包括主键ROW_ID
	 *
	 * @param connection
	 * @param tableName
	 * @return
	 * @throws SQLException
	 */
	private static String getFields(Connection connection, String tableName,String scheam) throws SQLException {
		long start = System.currentTimeMillis();
		StringBuilder fields = new StringBuilder();
		String queryFieldsSql = "SELECT COLNAME FROM SYSCAT.COLUMNS a WHERE a.TABNAME= '" + tableName+"' AND a.TABSCHEMA='"+scheam+"'";
		preparedStatement = connection.prepareStatement(queryFieldsSql);
		resultSet = preparedStatement.executeQuery();
		while (resultSet.next()) {
			if (!"ROW_ID".equals(resultSet.getString(1))) {
				fields.append(resultSet.getString(1) + ", ");
			}
		}
		long end = System.currentTimeMillis();
		System.out.println("获取表字段时间: " + (end - start) + "毫秒");
		if(fields.length()<2){
		    return "";
        }
		return fields.substring(0, fields.length() - 2);
	}

	/**
	 * 查询源表数据
	 *
	 * @param connection
	 * @param sql
	 * @param tableName
	 * @return
	 * @throws SQLException
	 */
	private static List<String> getFromData(Connection connection, String sql, String tableName) throws SQLException {
		long start = System.currentTimeMillis();
		List<String> resultList = new ArrayList<String>();
		StringBuilder resultData = null;
		preparedStatement = connection.prepareStatement(sql);
		resultSet = preparedStatement.executeQuery();
		int col = resultSet.getMetaData().getColumnCount();
		while (resultSet.next()) {
			resultData = new StringBuilder();
			for (int i = 2; i <= col; i++) {
				if (StringUtils.isNotBlank(resultSet.getString(i))) {
					if (resultSet.getString(i).contains("\\")) {
						resultData.append("'" + resultSet.getString(i).replace("'", "\'").replace("\"", "\\\"") + "',");
					} else {
						resultData
								.append("\"" + resultSet.getString(i).replace("'", "\'").replace("\"", "\\\"") + "\",");
					}
				} else {
					resultData.append("NULL,");
				}
			}
			resultList.add(resultData.substring(0, resultData.length() - 1));
		}
		long end = System.currentTimeMillis();
		System.out.println("查询源表数据时间: " + (end - start) + "毫秒, 表名: " + tableName);
		return resultList;
	}
	/**
	 * 批量保存数据，成功的提交，失败的回滚
	 *
	 * @param connection
	 * @param tableName
	 * @param fields
	 * @param batchCount
	 * @param fromList
	 * @throws SQLException
	 */
	private static void saveDataBatch(Connection connection, String tableName, String fields, int batchCount,
			List<String> fromList) throws SQLException {
		for (int i = 0; i < fromList.size() / batchCount + 1; i++) {
			long start = System.currentTimeMillis();
			StringBuilder toSql = new StringBuilder();
			toSql.append("INSERT IGNORE INTO ");
			toSql.append(tableName.toUpperCase());
			toSql.append("(" + fields + ")");
			toSql.append(" VALUES");
			List<String> subFromList = fromList.subList(i * batchCount,
					i < fromList.size() / batchCount ? (i + 1) * batchCount : fromList.size());
			if (subFromList != null && subFromList.size() > 0) {
				for (String fromData : subFromList) {
					toSql.append("(" + fromData + "),");
				}
				int result =0;
				try{
					result = executeUpdate(connection, toSql.substring(0, toSql.length() - 1));
				}catch(Exception e){
                   // FileUtil.writeLines(Arrays.asList(new String[]{toSql.toString()}),filePath,"utf-8");
				   System.out.println(e);
				}
				  System.out.println("mysql实际插入条数; "+result);
				if(0==result){
                    System.out.println(toSql.toString());
					writeErrorToFile(filePath, tableName.toUpperCase());
                    //FileUtil.writeLines(Arrays.asList(new String[]{toSql.toString()}),filePath,"utf-8");
				}

			}
			long end = System.currentTimeMillis();
			System.out.println("批量保存数据使用时间: " + (end - start) + "毫秒");
		}
	}

	/**
	 * 更新数据
	 *
	 * @param connection
	 * @param sql
	 * @return
	 * @throws SQLException
	 */
	private static int executeUpdate(Connection connection, String sql) throws SQLException {
		preparedStatement = connection.prepareStatement(sql);
		return preparedStatement.executeUpdate();
	}

	// 保存数据
	public static void writeErrorToFile(String path, String tableName) {
		try {
			File file = new File(path);
			// 文件不存在时候，主动穿件文件。
			if (!file.exists()) {
				file.createNewFile();
			}
			FileWriter fw = new FileWriter(file, false);
			BufferedWriter bw = new BufferedWriter(fw);
			bw.newLine();
			bw.write(tableName);
			bw.close();
			fw.close();
		} catch (Exception e) {
			System.out.println("保存异常表异常");
		}
	}
}
